package net.bwie.realtime.jtp.ods.job;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.common.utils.MysqlCdcUtil;
import net.bwie.realtime.jtp.ods.function.CdcDataEtlMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class JtpCdcMallDataJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);

        DataStream<String> cdcStream = cdcMysqlSource(env);
        // cdcStream.print("cdcStream");

        DataStream<String> etlStream = cdcStream.map(new CdcDataEtlMapFunction());
        // etlStream.print("etlStream");

        KafkaUtil.producerKafka(etlStream, "jtp_mall_topic");

        env.execute("JtpCdcMallDataJob");
    }

    private static DataStream<String> cdcMysqlSource(StreamExecutionEnvironment env) {
        String[] dimList = new String[]{"activity_info"
                , "activity_rule"
                , "base_category1"
                , "base_category2"
                , "base_category3"
                , "base_dic"
                , "base_province"
                , "base_region"
                , "base_trademark"
                , "coupon_info"
                , "sku_info"
                , "spu_info"
                , "user_info"};
        DataStream<String> dimStream = MysqlCdcUtil.cdcMysqlInitial(env, "jtp_mall", dimList);

        String[] factsList = new String[]{"activity_sku"
                , "cart_info"
                , "comment_info"
                , "coupon_range"
                , "coupon_use"
                , "favor_info"
                , "order_detail"
                , "order_detail_activity"
                , "order_detail_coupon"
                , "order_info"
                , "order_refund_info"
                , "order_status_log"
                , "payment_info"
                , "refund_payment"
                , "tbl_example"};
        DataStream<String> factsStream = MysqlCdcUtil.cdcMysqlEarliest(env, "jtp_mall", factsList);
        return dimStream.union(factsStream);
    }
}
